#coding:utf-8
from datetime import datetime
from kafka import *
from kafka import TopicPartition
import json

class ProducerHandler(object):
    def __init__(self,topic,hosts,api_version=(0,8,1),username=None,password=None):
        self.topic = topic
        kafka_args = {"api_version":api_version,"bootstrap_servers":hosts,"request_timeout_ms":7000}
        kafka_args = {"bootstrap_servers":hosts,"request_timeout_ms":7000}
        if username and password:
            kafka_args["security_protocol"] = 'SASL_PLAINTEXT'
            kafka_args["sasl_plain_username"] = username
            kafka_args["sasl_plain_password"] = password
            kafka_args["sasl_mechanism"] = 'PLAIN'
        self.producer = KafkaProducer(**kafka_args)

    def write(self,msg):
        return self.producer.send(self.topic, msg)

    def write2partititon(self,msg,partition):
        return self.producer.send(self.topic, key = partition, value = msg)

    def close(self):
        self.producer.close()


class SysProducerHandler(ProducerHandler):
    def __init__(self,topic,hosts, api_version = (0,10,1),username='sinaRecmd',password='0dd844e301d3671cd30aee0e8e0fe732'):
        super(SysProducerHandler, self).__init__(topic, hosts=hosts, username=username, password=password, api_version=api_version)


if __name__ == "__main__":
    log_producer = SysProducerHandler(topic="test", hosts=["10.13.40.138:9092"],username=None,password=None)
    #log_producer = SysProducerHandler(topic="test", hosts=["localhost:9092"],username=None,password=None)

    data = {"test":10}
    import time
    st = time.time()
    for _ in range(1000):
        data = {"test":time.time()}
        #ret = log_producer.write(json.dumps(data,ensure_ascii=False).encode('utf-8'))
        log_producer.write(json.dumps(data,ensure_ascii=False).encode('utf-8'))
        #log_producer.write("idrest".encode('utf-8'))
        #print(ret.get(timeout=1))
    print(1000*(time.time()-st))

